package com.example.kafka.service;

import com.example.kafka.config.KafkaConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.stereotype.Service;

import java.util.Properties;

/**
 * 关键业务 retries=0 Service
 */
@Service
public class RetriesLeakService {
    public void send1(String msg) {
        // 关键业务 retries=0，临时故障直接丢失消息
        Properties props = KafkaConfig.getProducerProps();
        props.put("retries", "0");
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        producer.send(new ProducerRecord<>("topic_retries", msg));
        // 未关闭 producer
    }

    public void send2(String msg) {
        // retries=Integer.MAX_VALUE，极易资源耗尽
        Properties props = KafkaConfig.getProducerProps();
        props.put("retries", String.valueOf(Integer.MAX_VALUE));
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        producer.send(new ProducerRecord<>("topic_retries", msg));
        // 未关闭 producer
    }

    public void send3(String msg) {
        // 高重试+低超时，重试实际无效
        Properties props = KafkaConfig.getProducerProps();
        props.put("retries", "10");
        props.put("request.timeout.ms", "10");
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        producer.send(new ProducerRecord<>("topic_retries", msg));
        // 未关闭 producer
    }

    public void send4(String msg) {
        // 不区分错误类型，盲目高重试
        Properties props = KafkaConfig.getProducerProps();
        props.put("retries", "20");
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        producer.send(new ProducerRecord<>("topic_retries", msg));
        // 未关闭 producer
    }

    public void send5(String msg) {
        // 未设置 delivery.timeout.ms，重试时间无上限
        Properties props = KafkaConfig.getProducerProps();
        props.put("retries", "10");
        // delivery.timeout.ms 未设置
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        producer.send(new ProducerRecord<>("topic_retries", msg));
        // 未关闭 producer
    }
}
